/*
 ***************************************************************************************
 * All rights Reserved
 * @Title:  AbstractMessageConsumer.java   
 * @Package com.rtf.repair.mq.common
 * @Description: 消费处理消息默认抽象实现，针对不同的mq 实现获取特殊信息以兼容不同的处理，如rocket
 * 				 具有tag 信息，其它mq可能会以其它形式存储（kafka放入header里）；针对不同mq默认实现
 * 				对应的消息体解析获取body和tag,如  onMessage(ConsumerRecord<String, byte[]> record)
 * 				解析之后执行onMessage(String tag, byte[] message)
 * 				
 * @author: sfl
 * @date:   2020-10-22 13:05:07   
 * @version V1.0 
 * @Copyright: 2020 FOSONG. All rights reserved. 
 * 注意：本内容仅限于公司内部使用，禁止外泄以及用于其他的商业目
 *  ---------------------------------------------------------------------------------- 
 * 文件修改记录
 *     文件版本：         修改人：             修改原因：
 ***************************************************************************************
 */
package com.rtf.framework.mq.common;

import java.util.List;
import java.util.stream.Collectors;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.rocketmq.common.message.MessageExt;

/**
 *  
 * @Description: 消费处理消息默认抽象实现，针对不同的mq 实现获取特殊信息以兼容不同的处理，如rocket
 * 				  具有tag 信息，其它mq可能会以其它形式存储（kafka放入header里）；针对不同mq默认实现
 * 				  对应的消息体解析获取body和tag,如  onMessage(ConsumerRecord<String, byte[]> record)
 * 				  解析之后执行onMessage(String tag, byte[] message)
 * @ClassName:  AbstractMessageConsumer
 * @author: sfl
 * @date:   2020-10-22 13:11:03
 * @since:  v1.0
 */
public abstract class AbstractBatchMessageConsumer implements BatchMessageConsumer {

	/* (non-Javadoc)
	 * @see com.rtf.repair.mq.common.MessageConsumer#onMessage(org.apache.kafka.clients.consumer.ConsumerRecord)
	 */
	@Override
	public boolean onMessageKafka(List<ConsumerRecord<String, byte[]>> records) {
		List<byte[]> msgList = records.stream().map(ConsumerRecord::value).collect(Collectors.toList());
		return this.onMessage(msgList);
	}

	/* (non-Javadoc)
	 * @see com.rtf.repair.mq.common.MessageConsumer#onMessage(org.apache.rocketmq.common.message.MessageExt)
	 */
	@Override
	public boolean onMessageRocket(List<MessageExt> records) {

		List<byte[]> msgList = records.stream().map(MessageExt::getBody).collect(Collectors.toList());
		return this.onMessage(msgList);

	}

}
